import com.arcadedb.database.Database;
import com.arcadedb.database.DatabaseFactory;
import com.arcadedb.graph.Vertex;
import com.arcadedb.query.sql.executor.ResultSet;
import com.arcadedb.schema.Schema;
import com.arcadedb.schema.VertexType;
import nju.hjh.arcadedb.timeseries.DataType;
import nju.hjh.arcadedb.timeseries.TSUpdateStrategy;
import nju.hjh.arcadedb.timeseries.datapoint.StringDataPoint;
import nju.hjh.arcadedb.timeseries.TimeseriesEngine;
import nju.hjh.arcadedb.timeseries.exception.DuplicateTimestampException;
import nju.hjh.arcadedb.timeseries.exception.TimeseriesException;
import indi.hjhk.csv.CSVReader;
import indi.hjhk.log.Logger;

import java.io.IOException;
import java.util.ArrayList;

public class DataTransTimeseries {
    public static void main(String[] args) throws CSVReader.CSVFormatException, IOException {
        Logger logger = Logger.getPureLogger("TSTrans");
        logger.disableAutoFlush();
        DatabaseFactory dbf = new DatabaseFactory("./databases/carTsTest");
        Database database;
        if (!dbf.exists()) {
            database = dbf.create();
        } else {
            database = dbf.open();
        }

        final String vertexType = "car";

        database.begin();
        if (!database.getSchema().existsType(vertexType)) {
            VertexType typeCar = database.getSchema().createVertexType(vertexType, 32);
            typeCar.createProperty("carId", "STRING").createIndex(Schema.INDEX_TYPE.LSM_TREE, true);
        }
        database.commit();

        TimeseriesEngine ts = new TimeseriesEngine(database);

        long totalSuccess = 0, totalFailed = 0, totalCsvElapsed = 0, totalArcadeElapsed = 0, totalTimeseriesElapsed = 0, totalCommitElapsed = 0;

        ArrayList<String> csvFiles = new ArrayList<>();
        csvFiles.add("/hdd/users/hjh/test.csv");

        final int commitSize = 1000, statSize = 300000;
        logger.logOnStdout("commit size= %d, stat size= %d", commitSize, statSize);

        for (String csvFilePath : csvFiles) {
            long offset = 0, success = 0, failed = 0;

            CSVReader csvReader = new CSVReader(csvFilePath);
            csvReader.readHeader();

            long startTime, csvElapsed = 0, arcadeElapsed = 0, timeseriesElapsed = 0, commitElapsed = 0;
            long subCsvElapsed = 0, subArcadeElapsed = 0, subTimeseriesElapsed = 0, subCommitElapsed = 0;

            logger.logOnStdout("handling data within file %s:", csvFilePath);

            int successCount = 0, failedCount = 0, subSuccessCount = 0, subFailedCount = 0;

            try {
                DataType stationIdType = new DataType(DataType.BaseType.STRING, 20);
                DataType passIdType = new DataType(DataType.BaseType.STRING, 40);

                ts.begin();

                while (csvReader.next()) {
                    startTime = System.currentTimeMillis();
                    String carId = csvReader.getString("VLP") + "_" + csvReader.getString("VLPC");
                    long timestamp = csvReader.getLong("TIME");
                    String passId = csvReader.getString("PASSID");
                    String stationId = csvReader.getString("STATIONID");

                    subCsvElapsed += System.currentTimeMillis() - startTime;
                    startTime = System.currentTimeMillis();

                    String sql = "select from " + vertexType + " where carId='" + carId + "'";
                    ResultSet arcRs;
                    try {
                        arcRs = database.query("SQL", sql);
                    } catch (Exception e) {
                        logger.logOnStderr("failed to parse sql statement: %s, data point skipped", sql);
                        subFailedCount++;
                        continue;
                    }

                    Vertex car;
                    if (arcRs.hasNext()) {
                        car = arcRs.next().getVertex().orElseGet(() -> database.newVertex(vertexType).set("carId", carId).save());
                    } else {
                        car = database.newVertex(vertexType).set("carId", carId).save();
                    }

                    subArcadeElapsed += System.currentTimeMillis() - startTime;
                    startTime = System.currentTimeMillis();

                    // ignore if timestamp exists
                    try{
                        ts.insertDataPoint(car, "stationId", stationIdType, new StringDataPoint(timestamp, stationId), TSUpdateStrategy.ERROR);
                        ts.insertDataPoint(car, "passId", passIdType, new StringDataPoint(timestamp, passId), TSUpdateStrategy.ERROR);
                        subSuccessCount++;
                    }catch (DuplicateTimestampException e){
                        // existing timestamp
                        subFailedCount++;
                    }

                    subTimeseriesElapsed += System.currentTimeMillis() - startTime;

                    if (subSuccessCount >= commitSize) {
                        startTime = System.currentTimeMillis();
                        ts.commit();
                        subCommitElapsed += System.currentTimeMillis() - startTime;

                        successCount += subSuccessCount;
                        failedCount += subFailedCount;
                        subSuccessCount = 0;
                        subFailedCount = 0;

                        ts.begin();
                    }

                    if (successCount >= statSize){
                        csvElapsed += subCsvElapsed;
                        arcadeElapsed += subArcadeElapsed;
                        timeseriesElapsed += subTimeseriesElapsed;
                        commitElapsed += subCommitElapsed;
                        success += successCount;
                        failed += failedCount;
                        offset += successCount + failedCount;

                        logger.logOnStdout("\tsuccessfully committed %d data (failed: %d, end index= %d), timeSpent: CSVReader: %dms + ArcadeDB: %dms + ArcadeDB Timeseries: %dms + commit: %dms = %dms", successCount, failedCount, offset - 1, subCsvElapsed, subArcadeElapsed, subTimeseriesElapsed, subCommitElapsed, subCsvElapsed + subTimeseriesElapsed + subArcadeElapsed + subCommitElapsed);

                        successCount = 0;
                        failedCount = 0;
                        subCsvElapsed = 0;
                        subTimeseriesElapsed = 0;
                        subArcadeElapsed = 0;
                        subCommitElapsed = 0;
                    }
                }
                startTime = System.currentTimeMillis();
                ts.commit();
                subCommitElapsed += System.currentTimeMillis() - startTime;

                successCount += subSuccessCount;
                failedCount += subFailedCount;

                csvElapsed += subCsvElapsed;
                arcadeElapsed += subArcadeElapsed;
                timeseriesElapsed += subTimeseriesElapsed;
                commitElapsed += subCommitElapsed;
                success += successCount;
                failed += failedCount;
                offset += successCount + failedCount;

                logger.logOnStdout("\tsuccessfully committed %d data (failed: %d, end index= %d), timeSpent: CSVReader: %dms + ArcadeDB: %dms + ArcadeDB Timeseries: %dms + commit: %dms = %dms", successCount, failedCount, offset - 1, subCsvElapsed, subArcadeElapsed, subTimeseriesElapsed, subCommitElapsed, subCsvElapsed + subTimeseriesElapsed + subArcadeElapsed + subCommitElapsed);

                totalCsvElapsed += csvElapsed;
                totalArcadeElapsed += arcadeElapsed;
                totalTimeseriesElapsed += timeseriesElapsed;
                totalCommitElapsed += commitElapsed;
                totalSuccess += success;
                totalFailed += failed;

            } catch (TimeseriesException e) {
                ts.rollback();
                database.close();
                logger.logOnStderr("Problem occurred when transferring from csv file %s to ArcadeDB at offset %d (+%d)", csvFilePath, offset, successCount);
                logger.flush();
                throw new RuntimeException(e);
            }

            logger.logOnStdout(String.format("Transferred from csv file %s to ArcadeDB(success: %d, failed: %d), CSV: %dms + ArcadeDB: %dms + ArcadeDB Timeseries: %dms + commit: %dms = %dms", csvFilePath, success, failed, csvElapsed, arcadeElapsed, timeseriesElapsed, commitElapsed, csvElapsed + arcadeElapsed + timeseriesElapsed + commitElapsed));
        }
        logger.logOnStdout(String.format("Total: Transferred from %d csv file(success: %d, failed: %d) to ArcadeDB, CSV: %dms + ArcadeDB: %dms + ArcadeDB Timeseries: %dms + commit: %dms = %dms", csvFiles.size(), totalSuccess, totalFailed, totalCsvElapsed, totalArcadeElapsed, totalTimeseriesElapsed, totalCommitElapsed, totalCsvElapsed + totalArcadeElapsed + totalTimeseriesElapsed + totalCommitElapsed));
        database.close();
        logger.flush();
    }
}
